package com.base.data.elasticsearch.autoconfig.conf;


import com.base.data.elasticsearch.core.BaseElasticsearchTemplate;
import com.base.data.elasticsearch.core.BaseRestHighLevelClient;
import com.base.data.elasticsearch.util.MiddleLevelRestClient;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

@Configuration
@EnableConfigurationProperties({BaseElasticsearchProperties.class})
public class BaseElasticsearchRestClientAutoconfigure {
    final static String ALIYUN = "aliyun";
    final static String SCHEMA = "http";

    @Autowired
    BaseElasticsearchProperties elasticsearchProperties;

    @Bean
    RestClient baseRestClient() {
        return getRestClient(elasticsearchProperties);
    }

    @Bean
    MiddleLevelRestClient middleLevelRestClient(@Autowired @Qualifier(value = "baseRestClient") RestClient client) {
        return elasticsearchProperties.getHttpBufferLimit() > 0 ? new MiddleLevelRestClient(client,elasticsearchProperties.getHttpBufferLimit()) : new MiddleLevelRestClient(client);
    }

    @Bean
    BaseRestHighLevelClient baseRestHighLevelClient(@Autowired @Qualifier(value = "baseRestClient") RestClient client,
                                                    @Autowired @Qualifier(value = "middleLevelRestClient") MiddleLevelRestClient middleLevelRestClient) {
        return new BaseRestHighLevelClient(client,middleLevelRestClient);
    }

    @Bean
    public BaseElasticsearchTemplate baseElasticsearchTemplate(@Autowired @Qualifier(value = "baseRestHighLevelClient") BaseRestHighLevelClient client,
                                                               @Autowired @Qualifier(value = "baseRestClient") RestClient restClient) {
        final BaseElasticsearchTemplate elasticsearchTemplate = new BaseElasticsearchTemplate(client, restClient);
        elasticsearchTemplate.setAutoMapping(elasticsearchProperties.getAutoMapping());
        return elasticsearchTemplate;
    }

    private RestClient getRestClient(BaseElasticsearchProperties properties) {
        RestClient restClient = null;
        if (ALIYUN.equals(properties.getType())) {
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY,
                    new UsernamePasswordCredentials(properties.username, properties.pass));

            restClient = RestClient.builder(new HttpHost(properties.getClusterNodes().split(":")[0], 9200))
                    .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                        @Override
                        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                            return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        }
                    })
                    .build();
        } else {
            List<HttpHost> list = Arrays.stream(properties.getClusterNodes().split(",")).map(x -> {
                String[] arr = x.split(":");
                String ip = arr[0];
                int port = 9200;
                return new HttpHost(ip, port, SCHEMA);
            }).collect(Collectors.toList());
            restClient = RestClient.builder(list.toArray(new HttpHost[list.size()])).setMaxRetryTimeoutMillis(30000
            ).build();
        }
        return restClient;
    }
}

